package outlierDetector;

import data.DataPoint;
import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.transforms.Combine.CombineFn;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Serializable;
import java.util.ArrayList;

/**
 * Created by emily on 17-12-13.
 */
public class DynamicThresholdDetector extends CombineFn<DataPoint, DynamicThresholdDetector.Accum, ArrayList<DataPoint>>
{

    private static final Logger LOG = LoggerFactory.getLogger(DynamicThresholdDetector.class);
    private static int WINDOWSIZE;
    private static double PROBABILITY;

    public DynamicThresholdDetector(int windowsize, double probability){
        this.WINDOWSIZE = windowsize;
        this.PROBABILITY = probability;

    }

    @DefaultCoder(AvroCoder.class)
    public static class Accum implements Serializable {
        double sum = 0;
        int count = 0;
        ArrayList<DataPoint> dpList = new ArrayList<DataPoint>();

    }

    @Override
    public Accum createAccumulator() { return new Accum(); }

    @Override
    public Accum addInput(Accum accum, DataPoint dp) {
        accum.sum += dp.value;
        accum.count++;
        accum.dpList.add(dp);
        return accum;
    }

    @Override
    public Accum mergeAccumulators(Iterable<Accum> accums) {
        Accum merged = createAccumulator();
        int i = 0;
        for (Accum accum : accums) {
            merged.sum += accum.sum;
            merged.count += accum.count;
            i++;
            for(DataPoint dp: accum.dpList){
                merged.dpList.add(dp);
            }

        }
        System.out.println("accums num when merged: " + i );
        System.out.println("merged dpList length: " + merged.dpList.size());
        return merged;
    }

    @Override
    public ArrayList<DataPoint> extractOutput(Accum accum) {
        System.out.println("dynamic threshold count: " + accum.count);
        //System.out.println(accum.dp.timestamp);
        System.out.println("extracctOutput accum dp: ");
        /*for(DataPoint dp: accum.dpList){
            System.out.println(dp.timestamp + "," + dp.value);
        }*/
        ArrayList<DataPoint> dpList = new ArrayList<DataPoint>();
        if(accum.count == WINDOWSIZE){
            double thres1 = 0.0;
            double thres2 = 0.0;
            double sum1 = 0.0;
            double sum2 = 0.0;
            int count1 = 2*WINDOWSIZE/3;
            int pos = WINDOWSIZE/3;

            for(int i = 0; i<count1; i++){
                sum1 += accum.dpList.get(i).value;
            }
            for(int i = pos; i<WINDOWSIZE; i++){
                sum2 += accum.dpList.get(i).value;
            }
            double threshold = -(sum1/count1 + sum2/count1) * Math.log(PROBABILITY);
            System.out.println("outlier threshold: " + threshold);

            for (int i = pos; i<count1; i++){
                if(accum.dpList.get(i).value > threshold){
                    dpList.add(accum.dpList.get(i));
                }
            }
            System.out.println("outlier num: " + dpList.size());


        }


        return dpList;
    }

}
